
package org.apache.solr.cloud;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.List;
import java.util.Properties;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.common.cloud.ZooKeeperException;
import org.apache.solr.core.SolrResourceLoader;
import org.apache.zookeeper.KeeperException;

/**
 * ResourceLoader that works with ZooKeeper.
 */
public class ZkSolrResourceLoader extends SolrResourceLoader {

    private final String collectionZkPath;
    private ZkController zkController;

    public ZkSolrResourceLoader(String instanceDir, String collection, ZkController zooKeeperController) {
        super(instanceDir);

        this.zkController = zooKeeperController;
        collectionZkPath = ZkController.CONFIGS_ZKNODE + "/" + collection;
    }

    /**
     * This loader will first attempt to load resources from ZooKeeper, but if not found
     * will delegate to the context classloader when possible,
     * otherwise it will attempt to resolve resources using any jar files found in
     * the "lib/" directory in the specified instance directory.
     */
    public ZkSolrResourceLoader(String instanceDir, String collection, ClassLoader parent, Properties coreProperties, ZkController zooKeeperController) {
        super(instanceDir, parent, coreProperties);

        this.zkController = zooKeeperController;
        collectionZkPath = ZkController.CONFIGS_ZKNODE + "/" + collection;
    }

    /**
     * Opens any resource by its name. By default, this will look in multiple
     * locations to load the resource: $configDir/$resource from ZooKeeper.
     * It will look for it in any jar
     * accessible through the class loader if it cannot be found in ZooKeeper.
     * Override this method to customize loading resources.
     *
     * @return the stream for the named resource
     */
    @Override
    public InputStream openResource(String resource) throws IOException {

        InputStream is = null;
        String file = collectionZkPath + "/" + resource;
        try {
            if(zkController.pathExists(file)) {
                byte[] bytes = zkController.getZkClient().getData(collectionZkPath + "/" + resource, null, null, true);
                return new ByteArrayInputStream(bytes);
            }
        }
        catch(KeeperException | InterruptedException e) {
            throw new IOException("Error opening " + file, e);
        }
        try {
            // delegate to the class loader (looking into $INSTANCE_DIR/lib jars)
            is = classLoader.getResourceAsStream(resource);
        }
        catch(Exception e) {
            throw new IOException("Error opening " + resource, e);
        }
        if(is == null) {
            throw new IOException("Can't find resource '" + resource + "' in classpath or '" + collectionZkPath + "', cwd=" + System.getProperty("user.dir"));
        }
        return is;
    }

    @Override
    public String getConfigDir() {
        throw new ZooKeeperException(ErrorCode.SERVER_ERROR, "ZkSolrResourceLoader does not support getConfigDir() - likely, what you are trying to do is not supported in ZooKeeper mode");
    }

    @Override
    public String[] listConfigDir() {

        List<String> list;
        try {
            list = zkController.getZkClient().getChildren(collectionZkPath, null, true);
        }
        catch(InterruptedException e) {
            // Restore the interrupted status
            Thread.currentThread().interrupt();
            log.error("", e);
            throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "", e);
        }
        catch(KeeperException e) {
            log.error("", e);
            throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "", e);
        }
        return list.toArray(new String[0]);
    }

    public String getCollectionZkPath() {
        return collectionZkPath;
    }
}
